#!/bin/bash

function task_job() {
jdbcUrl="$1"
db="$2"
mysqlTable="$3"
hiveTable="$4"
column="${5}"
sub="${6}"
threadNumber="${7}"
typed="${8}"
dt="{{ execution_date | cst_ds }}"
nextdt="{{ execution_date | date_add(1) | cst_ds }}"
#dt="{{ execution_date | date_add(-5) | cst_ds }}"
#nextdt="{{ execution_date | date_add(-4) | cst_ds }}"
env="{{ var.value.env_sync }}"
#账户和密码新加的实例和旧实例是一样的
username='{{ var.json.mysql_spmi.username }}'
password='{{ var.json.mysql_spmi.password }}'

json='{
"reader":{
"connect":{
"url":"'"${jdbcUrl}"'",
"username":"'"${username}"'",
"password":"'"${password}"'",
"driver":"com.mysql.cj.jdbc.Driver"
},
"dbtype":"mysql",
"tableName":"'"${db}.${mysqlTable}"'",
"subTableList":"'"${sub}"'",
"where":"'"${column} between str_to_date('${dt} 00:00:00', '%Y-%m-%d %H:%i:%s') and str_to_date('${nextdt} 00:00:00', '%Y-%m-%d %H:%i:%s')"'",
"query":"'""'",
"splitColumn":"'"${column}"'",
"equalitySectioning":0,
"containsnull":0,
"fetchsize":"1024",
"threadNumber":0
},
"channel":{
"filterAbnormalCharacter":0
},
"writer":{
"dag":"spmi_ods",
"task_id":"spmi_ods__spmi_piece_bill",
"dbtype":"hive",
"tableName":"'"${hiveTable}"'",
"database":"spmi_ods",
"writeMode": "'"${typed}"'",
"partitionColumn":"dt",
"partitionValue":"'"${dt}"'"},
"settting":{
"env":"'"${env}"'"}
}'

/usr/hdp/3.1.4.0-315/spark2/bin/spark-submit \
--master yarn \
--deploy-mode cluster \
--driver-memory 18G \
--driver-cores 4 \
--executor-memory 8G \
--executor-cores 2 --queue pro  \
--num-executors ${threadNumber}  \
--conf spark.driver.maxResultSize=6g \
--class com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver \
hdfs:///scheduler/jms/spark/sync/mysql/spark_sync.jar  "$json"
echo ${json} >> /jms_ods/spmi_piece_bill.log
echo "任务执行成功: ${dt} ${hiveTable} ${db} ${sub}"
}

mTab="spmi_piece_bill"
mTab_new="spmi_piece_bill_new"
hTab="spmi_piece_bill"
col='update_time'
#jdbcUrl1='{{ var.json.mysql_spmi.db01 }}'
#jdbcUrl2='{{ var.json.mysql_spmi.db02 }}'

#2022-01-24日任务添加新的实例
jdbcUrl11='{{ var.json.mysql_piecebill.db11 }}'
jdbcUrl13='{{ var.json.mysql_piecebill.db13 }}'
jdbcUrl15='{{ var.json.mysql_piecebill.db15 }}'

jdbcUrl21='{{ var.json.mysql_piecebill.db21 }}'
jdbcUrl23='{{ var.json.mysql_piecebill.db23 }}'
jdbcUrl25='{{ var.json.mysql_piecebill.db25 }}'

jdbcUrl31='{{ var.json.mysql_piecebill.db31 }}'
jdbcUrl33='{{ var.json.mysql_piecebill.db33 }}'
jdbcUrl35='{{ var.json.mysql_piecebill.db35 }}'

jdbcUrla1='{{ var.json.mysql_piecebill.dba1 }}'
jdbcUrla3='{{ var.json.mysql_piecebill.dba3 }}'
jdbcUrla5='{{ var.json.mysql_piecebill.dba5 }}'


#2022-01-24日新增新实例库同步
task_job  ${jdbcUrl11}  yl_piecebill_1 ${mTab} ${hTab} ${col} "[0,255]" 3  overwrite
task_job  ${jdbcUrl13}  yl_piecebill_3 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl15}  yl_piecebill_5 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &

wait

task_job  ${jdbcUrl21}  yl_piecebill_1 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl23}  yl_piecebill_3 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl25}  yl_piecebill_5 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &

wait

task_job  ${jdbcUrl31}  yl_piecebill_1 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl33}  yl_piecebill_3 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl35}  yl_piecebill_5 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &

wait

task_job  ${jdbcUrla1}  yl_piecebill_1 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrla3}  yl_piecebill_3 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrla5}  yl_piecebill_5 ${mTab} ${hTab} ${col} "[0,255]" 3  append    &

wait
# 新增表-new 2022-09-21
task_job  ${jdbcUrl11}  yl_piecebill_1 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl13}  yl_piecebill_3 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl15}  yl_piecebill_5 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &

wait
task_job  ${jdbcUrl21}  yl_piecebill_1 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl23}  yl_piecebill_3 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl25}  yl_piecebill_5 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &

wait

task_job  ${jdbcUrla1}  yl_piecebill_1 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrla3}  yl_piecebill_3 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrla5}  yl_piecebill_5 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &

wait

task_job  ${jdbcUrl31}  yl_piecebill_1 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl33}  yl_piecebill_3 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &
task_job  ${jdbcUrl35}  yl_piecebill_5 ${mTab_new} ${hTab} ${col} "[0,255]" 3  append    &


wait
echo "${hTab} 表所有任务执行成功! "
